package org.flink.marketing.sink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.flink.marketing.constant.EventTrackingConstant;
import org.flink.marketing.constant.SqlConstant;
import org.flink.marketing.domain.BaseApplicationInfoDO;
import org.flink.marketing.domain.BaseBlackListDO;
import org.flink.marketing.domain.EventItemDO;
import org.flink.marketing.domain.EventTypeDO;
import org.flink.marketing.util.CheckUtil;
import org.flink.marketing.util.NumberUtil;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.io.InputStream;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Date;

@Slf4j
public class ClickHouseSink extends RichSinkFunction<JSONObject> {
    private Connection connection;
    private PreparedStatement preparedStatement;
    private transient Jedis jedis;
    private static Properties properties;

    private static void init() {
        properties = new Properties();
        try(InputStream inputStream = ClickHouseSink.class.getResourceAsStream("/ck.properties")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        init();
        super.open(parameters);
        // 加载JDBC驱动
        Class.forName(properties.getProperty("DriverClass"));
        // 获取数据库连接
        connection = DriverManager.getConnection(properties.getProperty("DriverUrl"),properties.getProperty("User"),properties.getProperty("Password"));

    }

    @Override
    public void close() throws Exception {
        super.close();
        connection.close();
    }

    @Override
    public void invoke(JSONObject jsonValue, Context context) throws Exception {

        if (jsonValue == null || jsonValue.isEmpty()){
            log.info("jsonValue为空："+jsonValue);
            return;
        }
        log.info("ck开始："+JSONObject.toJSONString(jsonValue));

        //增加用户埋点数据处理
        //login_user_data 标识用户属性的事件
        if(jsonValue.toString().contains(SqlConstant.LOGIN_USER_DATA)){
            initUser(jsonValue);
            return;
        }

        StringBuffer sbSql = new StringBuffer();
        sbSql.append(SqlConstant.INSERT_INTO).append(SqlConstant.INSERT_TABLE);
        sbSql.append("(");
        List<String> tableColums = new ArrayList<>();
        List<String> fileds = new ArrayList<>();

        Iterator<String> it = jsonValue.keySet().iterator();
        //增加黑名单校验
//        if(checkIpEvent(it,jsonValue)){
//            return;
//        }

        while(it.hasNext()){
            // 获得key
            String key = it.next();
            String value = jsonValue.getString(key);
            if (key.startsWith("$")){
                key = key.substring(1,key.length());
            }
            if (EventTrackingConstant.LIB.equals(key)
                    || EventTrackingConstant.PROPERTIES.equals(key)){
                continue;
            }
            if (key.startsWith("_")){
                continue;
            }
            if (!CheckUtil.checkColumn(key)){
                continue;
            }
            if (fileds.contains(key)){
                continue;
            }
            if (EventTrackingConstant.TIME.equals(key)){
                continue;
            }

//            System.out.println("key: "+key+",value:"+value);
            sbSql.append(key).append(",");
            value = value == null?"":value;
            if (value.startsWith("$")){
                value = value.substring(1,value.length());
            }
            //如果入参time为空，用系统时间
            if (EventTrackingConstant.TIME.equals(key)){
                if(value == null || value.equals("")){
                    tableColums.add(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                }else{
                    tableColums.add(getFormatDate(value));
                }
            }else{
                tableColums.add(value);
            }

            fileds.add(key);

        }
        //lib节点属性
        JSONObject libJsonValue = jsonValue.getJSONObject(EventTrackingConstant.LIB);
        Iterator<String> itLib = libJsonValue.keySet().iterator();
        while(itLib.hasNext()){
            // 获得key
            String key = itLib.next();
            String value = libJsonValue.getString(key);
//            System.out.println("key: "+key+",value:"+value);
            if (key.startsWith("$")){
                key = key.substring(1,key.length());
            }
            if (key.equals(EventTrackingConstant.LIB)){
                key = key+"1";
            }
            if (!CheckUtil.checkColumn(key)){
                continue;
            }
            if (fileds.contains(key)){
                continue;
            }
            if (EventTrackingConstant.TIME.equals(key)){
                continue;
            }

            sbSql.append(key).append(",");
            value = value == null?"":value;
            if (value.startsWith("$")){
                value = value.substring(1,value.length());
            }
            tableColums.add(value);

            fileds.add(key);

        }
        //properties 节点属性
        JSONObject propertiesJsonValue = jsonValue.getJSONObject(EventTrackingConstant.PROPERTIES);
        Iterator<String> itProperties = propertiesJsonValue.keySet().iterator();
        while(itProperties.hasNext()){
            // 获得key
            String key = itProperties.next();
            String value = propertiesJsonValue.getString(key);
            if (key.startsWith("$")){
                key = key.substring(1,key.length());
            }
            if (!CheckUtil.checkColumn(key)){
                continue;
            }
            if (fileds.contains(key)){
                continue;
            }
            if (EventTrackingConstant.TIME.equals(key)){
                continue;
            }

//            System.out.println("key: "+key+",value:"+value);
            sbSql.append(key).append(",");
            value = value == null?"":value;
            if (value.startsWith("$")){
                value = value.substring(1,value.length());
            }
            tableColums.add(value);

            fileds.add(key);

        }
        if (!CheckUtil.checkTime(EventTrackingConstant.TIME,fileds)){
            tableColums.add(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        }
//        tableColums.add(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        tableColums.add(new SimpleDateFormat("yyyyMMdd").format(new Date()));
        if (!CheckUtil.checkTime(EventTrackingConstant.TIME,fileds)){
            sbSql.append("time,");
        }
//        sbSql.append("time,").append("dt");
        sbSql.append("dt");
        sbSql.append(")");
        //values 分隔
        sbSql.append(SqlConstant.INSERT_VALUE).append("(");

        int length = tableColums.size();
        int t=0;
        for (String s :tableColums){
            if (t+1 == length){
                sbSql.append("?");
            }else {
                sbSql.append("?").append(",");
            }
            t++;
        }
//        sbSql.append("?,").append("?");
        sbSql.append(")");

        String sql = sbSql.toString();
        log.info(sql);

        PreparedStatement preparedStatement = connection.prepareStatement(sql);

        for (int i = 0; i < length; i++) {
            preparedStatement.setString(i + 1, tableColums.get(i));
        }
        int ints = preparedStatement.executeUpdate();
//        preparedStatement.addBatch();

        long startTime = System.currentTimeMillis();
//        int[] ints = preparedStatement.executeBatch();
        connection.commit();
        long endTime = System.currentTimeMillis();
        log.info("批量插入完毕用时：" + (endTime - startTime) + " -- 插入数据 = " + ints);
    }

    private String getFormatDate(String value){
        Date date = null;
        if (NumberUtils.isNumber(value)){
            long times = new Long(value).longValue();
            date = new Date(times);
        }else{
            date = new Date();
        }

        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        String dateString = formatter.format(date);

        return dateString;

    }

    //校验埋点数据
    private  Boolean  checkIpEvent(Iterator<String> it,JSONObject jsonValue){
        Boolean flag=false;
        while(it.hasNext()){
            String key = it.next();
            String value = jsonValue.getString(key);
            if (key.startsWith("$")){
                key = key.substring(1,key.length());
            }
            //校验IP
            if(key.equals("IP")){
                //获取黑名单
                List<BaseBlackListDO> listDOS = JSON.parseArray(jedis.get(CheckUtil.BASEBLACKLIST_KEY),BaseBlackListDO.class);
                if(listDOS!=null&&listDOS.size()>0){
                    for(BaseBlackListDO  bo:listDOS){
                        if(value.equals(bo.getValue())){
                            flag=true;
                            break;
                        }
                    }
                }
            }
        }
        return flag;
//        //获取应用
//        List<BaseApplicationInfoDO> baseApplicationInfoDOList = JSON.parseArray(jedis.get(CheckUtil.BASEAPPLICATIONINFODOLIST_KEY),BaseApplicationInfoDO.class);
//        //校验应用code
//        if(key.equals("App_id")){
//            if(baseApplicationInfoDOList!=null&&baseApplicationInfoDOList.size()>0){
//                for(BaseApplicationInfoDO  bo:baseApplicationInfoDOList){
//                    if(value.equals(bo.getAppCode())){
//                        flag=false;
//                        break;
//                    }
//                }
//            }
//        }
//
//        //获取事件
//        List<EventTypeDO> eventTypeDOList = JSON.parseArray(jedis.get(CheckUtil.EVENTTYPEDOLIST_KEY),EventTypeDO.class);
//        if(key.equals("event")){
//            if(eventTypeDOList!=null&&eventTypeDOList.size()>0){
//                for(EventTypeDO  bo:eventTypeDOList){
//                    //同一个应用下的验证
//                    if(jsonValue.getString("App_id").equals(bo.getTeamCode())){
//                        if(value.equals(bo.getEventTypeName())){
//                            flag=false;
//                            break;
//                        }
//                    }
//                }
//            }
//
//        }
//
//        //获取事件属性
//        List<EventItemDO> eventItemDOList = JSON.parseArray(jedis.get(CheckUtil.EVENTITEMDOLIST_KEY),EventItemDO.class);
//        if(key.equals("event")){
//            if(eventItemDOList!=null&&eventItemDOList.size()>0){
//                for(EventItemDO  eventItemDO:eventItemDOList){
//                    if(jsonValue.getString("App_id").equals(eventItemDO.getTeamCode())){
//
//                    }
//
//                }
//            }
//        }

    }

    public  void initUser(JSONObject jsonValue) throws SQLException {

        //处理用户埋点属性分支
        //用户埋点数据报文格式
        //login_id、event_name、properties这个3个属性是固定的，properties里面的属性是动态的
        /*{
                "login_id": "123456789",
                "event_name": "login_user_data",
                "properties": {
                    "sex": "男",
                    "age": "30",
                    "name": "张三",
                    "application_ode":"xiaoxiangApp"  //应用编码
                 }
        }*/
        //login_user_data 标识用户属性的事件
        //用户属性只采集UI登录成功之后的数据，用户ID为唯一表示，对用户表数据进行增加，修改

        //第一步，查询用户ID数据是否存在
        String loginId=jsonValue.getString(SqlConstant.LOGINID);

        //用户ID没有不处理
        if(StringUtils.isEmpty(loginId)){
            return;
        }

        StringBuffer selectSql = new StringBuffer();
        selectSql.append(SqlConstant.SELECCTSQL).append(SqlConstant.WHERELOINGID).append(loginId);
        String sql = selectSql.toString();
        log.info("执行用户查询操作....................:"+sql);

        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        ResultSet resultSet = preparedStatement.executeQuery();

        List<String> tableColums = new ArrayList<>();
        List<String> fileds = new ArrayList<>();

        //更新
        if(resultSet.next()){

            //properties 节点属性
            JSONObject propertiesJsonValue = jsonValue.getJSONObject(EventTrackingConstant.PROPERTIES);
            Iterator<String> itProperties = propertiesJsonValue.keySet().iterator();

            StringBuffer s=new StringBuffer();
            while(itProperties.hasNext()){
                // 获得key
                String key = itProperties.next();
                String value = propertiesJsonValue.getString(key);
                value = value == null?"":value;
                tableColums.add(value);
                s.append(key).append(" = ").append(" ? ").append(",");
            }

            StringBuffer updateSql = new StringBuffer();
            updateSql.append(SqlConstant.UPDATESQL);

            updateSql.append(s.substring(0,s.length()-1)).append(SqlConstant.WHERELOINGID).append(loginId);

            String isql = updateSql.toString();
            log.info("用户数据更新SQL.......:"+isql);

            PreparedStatement updateConnection = connection.prepareStatement(isql);

            for (int i = 0; i < tableColums.size(); i++) {
                updateConnection.setString(i + 1, tableColums.get(i));
            }

            int ints = updateConnection.executeUpdate();
            long startTime = System.currentTimeMillis();
            connection.commit();
            long endTime = System.currentTimeMillis();
            log.info("更新用户数据完毕用时：" + (endTime - startTime) + " -- 插入数据 = " + ints);

        }

        //新增
        else{

            StringBuffer insertSql = new StringBuffer();
            insertSql.append(SqlConstant.INSERTSQL);
            insertSql.append("(");

            insertSql.append(SqlConstant.LOGINID).append(",");
            fileds.add(SqlConstant.LOGINID);
            tableColums.add(loginId);

            StringBuffer tsql=new StringBuffer();

            //properties 节点属性
            JSONObject propertiesJsonValue = jsonValue.getJSONObject(EventTrackingConstant.PROPERTIES);
            Iterator<String> itProperties = propertiesJsonValue.keySet().iterator();
            while(itProperties.hasNext()){
                // 获得key
                String key = itProperties.next();
                String value = propertiesJsonValue.getString(key);
                if (fileds.contains(key)){
                    continue;
                }

                tsql.append(key).append(",");
                value = value == null?"":value;
                tableColums.add(value);
                fileds.add(key);
            }

            insertSql.append(tsql.substring(0,tsql.length()-1)).append(")");
            insertSql.append(SqlConstant.INSERT_VALUE).append("(");
            int length = tableColums.size();
            int t=0;
            for (String s :tableColums){
                if (t+1 == length){
                    insertSql.append("?");
                }else {
                    insertSql.append("?").append(",");
                }
                t++;
            }
            insertSql.append(")");

            String isql = insertSql.toString();

            log.info("用户数据新增SQL.......:"+isql);

            PreparedStatement insertConnection = connection.prepareStatement(isql);

            for (int i = 0; i < length; i++) {
                insertConnection.setString(i + 1, tableColums.get(i));
            }

            int ints = insertConnection.executeUpdate();
            long startTime = System.currentTimeMillis();
            connection.commit();
            long endTime = System.currentTimeMillis();
            log.info("插入用户用户完毕用时：" + (endTime - startTime) + " -- 插入数据 = " + ints);
        }
    }
}
